package com.cctc.rds.raft.core.node;

import com.cctc.rds.raft.core.RaftCluster;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Raft 集群节点信息。
 */
public class Node {
    // 节点 ID
    private Integer id;
    // 所属角色：跟随者、候选者、领导者。
    private AtomicReference<NodeRole> role = new AtomicReference<>(NodeRole.Candidate);
    // 网络信息：ip、port
    private AtomicReference<NodeIpPort> ipPort = new AtomicReference<>(null);
    // 当前任期号
    private AtomicLong curTerm = new AtomicLong(0);
    // 已提交日志索引
    private AtomicLong commitLogIndex = new AtomicLong(0);
    // 投票器
    private final NodeVoterHolder voterHolder = new NodeVoterHolder();
    // 拉票计数器
    private final AtomicInteger voteNumber = new AtomicInteger(0);

    public Node(Integer id) {
        this.id = id;
    }

    public Integer getId() {
        return id;
    }

    public void setRole(NodeRole role) {
        this.role.set(role);
    }

    public NodeRole getRole() {
        return role.get();
    }

    public void setIpPort(NodeIpPort ipPort) {
        this.ipPort.set(ipPort);
    }

    public NodeIpPort getIpPort() {
        return ipPort.get();
    }

    public void initCurTerm(long term) {
        this.curTerm.compareAndSet(0, term);
    }

    public long getCurTerm() {
        return curTerm.get();
    }

    public void seekCommitLogIndex(long newCommitLogIndex) {
        this.commitLogIndex.set(newCommitLogIndex);
    }

    public long curCommitLogIndex() {
        return commitLogIndex.get();
    }

    public NodeVoterHolder getVoterHolder() {
        return voterHolder;
    }

    /**
     * 开始竞选时，重置拉票计数器。
     * - 将自身节点角色设置为候选人；
     * - 置新当前任期号；
     * - 根据新任期号获取投票器，并将第一票投给自己。
     */
    public boolean resetVoteNumber() {
        this.role.set(NodeRole.Candidate);
        this.curTerm.getAndIncrement();
        NodeVoter voter = voterHolder.get(getCurTerm());
        if (voter.lock(this.id)) {
            this.voteNumber.set(1);
            return true;
        }
        return false;
    }

    public void incVoteNumber() {
        this.voteNumber.getAndIncrement();
    }

    public int getVoteNumber() {
        return voteNumber.get();
    }

    /**
     * 竞选成功时，更新自身角色。
     * - 将自身角色设置为领导人；
     * - 更新自身当前任期号为全集群最新的任期号；
     * - 将其他节点设置为跟随者。
     */
    public void asLeader(long leaderTerm) {
        if (leaderTerm > getCurTerm()) {
            this.role.set(NodeRole.Leader);
            this.curTerm.set(leaderTerm);
            for (Node node : RaftCluster.getAllNodes()) {
                node.asFollower(leaderTerm);
            }
        }
    }

    /**
     * 竞选失败时，更新自身角色。
     * - 将自身角色设置为跟随者；
     * - 更新自身当前任期号，与领导人的任期号（全集群最新的任期号）保持一致。
     */
    public void asFollower(long leaderTerm) {
        if (leaderTerm > getCurTerm()) {
            this.role.set(NodeRole.Follower);
            this.curTerm.set(leaderTerm);
        }
    }

    @Override
    public boolean equals(Object object) {
        if (this == object) return true;
        if (object == null || getClass() != object.getClass()) return false;
        Node node = (Node) object;
        return Objects.equals(id, node.id);
    }

    @Override
    public int hashCode() {
        return Objects.hash(id);
    }

    @Override
    public String toString() {
        return "Node{" +
                "id=" + id +
                ", role=" + role +
                ", ipPort=" + ipPort +
                ", curTerm=" + curTerm +
                ", commitLogIndex=" + commitLogIndex +
                ", voterHolder=" + voterHolder +
                ", voteNumber=" + voteNumber +
                '}';
    }
}
